package com.hadoop.mr2.inputwordcount;

import java.io.IOException;
import java.io.InputStream;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.util.LineReader;

/**
 * 两个泛型类型对应的是输入key和输入value类型
 * @author ysq
 *
 */
public class WordCountRecordReader extends RecordReader<IntWritable,Text>{
	private FileSplit fs;
	private IntWritable key;
	private Text value;
	private LineReader reader;
	private int count;
	
	@Override
	public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
		fs=(FileSplit) split;
		//获取当前读取器处理的文件的路径path。（注意导包，导的是hadoop包）
		Path path=fs.getPath();
		Configuration conf=context.getConfiguration();
		FileSystem system=path.getFileSystem(conf);
		//获取当前所读取的文件的输入流
		InputStream in=system.open(path);
		//让行读取器里文件数据，所以最终目的是在初始化方法里，让行读取器有文件数据。
		reader=new LineReader(in);
		
		
	}
	
	/*
	 * ①此方法会被调用多次，如果返回true，则会被掉一次，直到返回false，停止调动
	 * 比如：for(int i=0;i<10;i++){
			return true;
		}
		return false;
		会调用10次
		
	 * ②nextKeyValue()每调用一次，getCurrentKey()和getCurrentValue()也会被跟着调用一次
	 * ③Mapper的输出key是通过getCurrentKey()这个方法传过去的
	 * ④Mapper的输出value是通过getCurrentValue()这个方法传的
	 */
	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
		key=new IntWritable();
		value=new Text();
		//临时变量，用于读取文件的每行内容
		Text tmp=new Text();
		int length= reader.readLine(tmp);
		if(length==0){
			//如果length=0,就证明所有行都已读完，文件处理完毕
			return false;
		}else{
			count++;
			//设置输入key是每行行号
			key.set(count);
			//设置输入value是每行内容
			value.set(tmp);
			
			return true;
		}
		
		
	}

	@Override
	public IntWritable getCurrentKey() throws IOException, InterruptedException {
		
		return key;
	}

	@Override
	public Text getCurrentValue() throws IOException, InterruptedException {
		
		return value;
	}

	@Override
	public float getProgress() throws IOException, InterruptedException {
		
		return 0;
	}

	@Override
	public void close() throws IOException {
		if(reader!=null)reader=null;
		
	}

}
